package com.shujia.core

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo01WordCount {
  def main(args: Array[String]): Unit = {
    // 创建Spark的环境
    val conf: SparkConf = new SparkConf() // 对Spark进行配置
    conf.setAppName("Demo01WordCount") // 设置程序名称
    conf.setMaster("local") // 设置Spark的运行方式，如果是本地 则设为local即可

    val sc: SparkContext = new SparkContext(conf)

    // 1、读取数据
    /**
     * RDD：在Spark中最核心的东西
     * 弹性分布式数据集
     * 先把它当成List去使用
     * 统一了编程模型
     */
    val lineRDD: RDD[String] = sc.textFile("spark/data/words.txt")

    val lineRDD2: RDD[String] = lineRDD.map(line => {
      println("执行了map方法")
      line
    })

    // 2、将每一个单词切分出来
    val wordsRDD: RDD[String] = lineRDD2.flatMap(_.split(","))

    // 3、按照每个单词进行分组
    val grpRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy(word => word)

    // 4、统计单词的数量
    val wordCntRDD: RDD[String] = grpRDD.map(kv => s"${kv._1},${kv._2.size}")

    // 5、打印
    wordCntRDD.foreach(println)

    // 判断输出路径是否存在，存在即删除
    val fileSystem: FileSystem = FileSystem.get(sc.hadoopConfiguration)
    val outPath: Path = new Path("spark/data/wc")
    if (fileSystem.exists(outPath)) {
      fileSystem.delete(outPath, true)
    }

    // 保存数据
    //    wordCntRDD.saveAsTextFile("spark/data/wc")

    // 链式调用 基本同Scala一致
    //    sc
    //      .textFile("spark/data/words.txt")
    //      .flatMap(_.split(","))
    //      .groupBy(word => word)
    //      .map(kv => s"${kv._1},${kv._2.size}")
    //      .foreach(println)

    //    // 不让程序自动停止
    //    while (true) {
    //      // 方便在本地查看WEB UI
    //    }
  }


}
